package sunyu.toolkit.kafka;

import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;

/**
 * kafka工具类
 *
 * @author 孙宇
 */
public class KafkaKit {

    private static final Logger logger = LoggerFactory.getLogger(KafkaKit.class);

    private static final Map<String, Producer> producerPool = new ConcurrentHashMap<>();//生产者池
    private static final Map<String, KafkaConsumerTemplate> consumerPool = new ConcurrentHashMap<>();//消费者池

    /**
     * 系统关闭前调用此方法
     */
    public static void destroy() {
        logger.info("销毁Kafka生产者和消费者开始");
        for (Map.Entry<String, Producer> entry : producerPool.entrySet()) {
            entry.getValue().close();
        }

        for (Map.Entry<String, KafkaConsumerTemplate> entry : consumerPool.entrySet()) {
            entry.getValue().setRunner(false);
        }
        logger.info("销毁Kafka生产者和消费者结束");
    }

    /**
     * 添加一个生产者
     *
     * @param name     生产者名称
     * @param producer 生产者模版
     */
    public static void addProducer(String name, KafkaProducerTemplate producer) {
        logger.info("添加Kafka Producer {}", name);
        if (producerPool.containsKey(name)) {
            logger.error("Kafka Producer {} 已存在!", name);
        } else {
            producerPool.put(name, producer.getProducer());
        }
    }

    /**
     * 获得生产者
     *
     * @param name 生产者名称
     * @return 生产者对象
     */
    public static Producer getProducer(String name) {
        return producerPool.get(name);
    }

    /**
     * 发送消息
     *
     * @param producerName 生产者名称
     * @param record       消息
     * @return 异步结果
     */
    public static Future send(String producerName,
                              ProducerRecord record) {
        return producerPool.get(producerName).send(record);
    }

    /**
     * 添加消费者
     *
     * @param name     消费者名称
     * @param consumer 消费者模版
     */
    public static void addConsumer(String name,
                                   KafkaConsumerTemplate consumer) {
        logger.info("添加Kafka Consumer {}", name);
        if (consumerPool.containsKey(name)) {
            logger.error("Kafka Consumer {} 已存在!", name);
        } else {
            consumerPool.put(name, consumer);
        }
    }

    /**
     * 获得消费者
     *
     * @param name 消费者名称
     * @return 消费者对象
     */
    public static KafkaConsumerTemplate getConsumer(String name) {
        return consumerPool.get(name);
    }

}
